/*
 * ClientAPI.java
 * --------------
 * Copyright (C) 2003-2007 tick-TS AG, All Rights Reserved
 *
 * This class encapsulates all client API functionality
 * and connection management
 *
 * $Id$
 *
 */

package de.tbmx.api;

import java.io.*;
import java.util.*;
import java.net.*;

public class ClientAPI 
{
	/** Generate a ClientAPI instance which connects to a locally-running TradeBase MX
	 * instance on the default port
	 */	
	public ClientAPI()
	{
		this("localhost",40411);
	}

	/** Create a ClientAPI instance which connects to a remote TradeBase MX instance on
	 * the specified host using the default port
	 * @param remoteHost Host to connect to
	 */	
	public ClientAPI(String remoteHost)
	{
		this(remoteHost,40411);
	}

	/** Connect to a remote TradeBase MX instance on the specified host and using the
	 * specified port
	 * @param remoteHost Host to connect to
	 * @param remotePort Port to connect to
	 */	
	public ClientAPI(String remoteHost, int remotePort)
	{
		this.remoteHost=remoteHost;
		this.remotePort=remotePort;
	}
	
	/** Connections will be automatically initiated whenever the first subscription or
	 * order is sent.  To ensure in advance that a connection has been initiated,
	 * call this method.
	 * 
	 */
	private boolean enablePositionUpdate = false;
	public void initiateConnection( boolean positionEnabled )
	{
		enablePositionUpdate = positionEnabled;
		checkThread();
	}
	
	/** Request that quotes be sent to the specified QuoteListener for the given isin 
	 * or symbol on the given exchange.  If level2 is true, both normal and market
	 * depth quotes will be sent.
	 * @param isinOrSymbol The ISIN or Symbol for which quotes are desired
	 * @param exchange The exchange for which quotes are desired
	 * @param level2 Whether market depth quotes should be sent to the QuoteListener
	 * @param ql The QuoteListener to which quotes should be sent
	 * @return A Quote object from which prices and information can be retrieved.  Note
	 * that this information might only be available some time later for instruments 
	 * which are being subscribed to for the first time.
	 */
	public Quote addQuoteListener(String isinOrSymbol, String exchange, boolean level2, QuoteListener ql)
	{
		synchronized(quotes)
		{
			Quote foundq=null;
			String key=isinOrSymbol+"."+exchange;
			foundq=quotesByISINExchange.get(key);
			if(foundq==null)
				foundq=quotesBySymbolExchange.get(key);
			if(foundq==null)
			{
				foundq=new Quote(isinOrSymbol,exchange);
				quotes.add(foundq);
				if(foundq.getISIN()!=null)
					quotesByISINExchange.put(key,foundq);
				else
					quotesBySymbolExchange.put(key,foundq);
			}
			safeWrite(foundq.getSubscriptionString(true,level2));
			foundq.addListener(ql, level2);
			return foundq;
		}
	}
	
	/** Request that quotes for the given Quote object no longer be sent to the given
	 * QuoteListener.
	 * @param q The Quote object for which quotes should no longer be sent
	 * @param ql The QuoteListener object to which quotes should no longer be sent
	 */
	public void removeQuoteListener(Quote q, QuoteListener ql)
	{
		boolean level2 = q.removeListener(ql);
		safeWrite(q.getSubscriptionString(false,level2));
		if (q.getL1Count()==0 && q.getL2Count()==0)
		{
			synchronized (quotes)
			{
				quotes.remove(q);
			}
		}
	}
	
	/** Return the number of InstrumentExchanges for which quotes are subscribed
	 * @return The number of InstrumentExchanges for which quotes are subscribed
	 */
	public int getQuoteCount()
	{
		return quotes.size();
	}

	public void addStatusListener(StatusListener sl)
	{
		synchronized(statusListeners)
		{
			statusListeners.add(sl);
			if(sl instanceof ExtendedStatusListener)
				extendedStatusListeners.add((ExtendedStatusListener)sl);
		}
	}
	
	PositionListener positionListener = null;
	
	public void addPositionListener( PositionListener pl ){
		positionListener = pl;
	}

	public void remStatusListener(StatusListener sl)
	{
		synchronized(statusListeners)
		{
			statusListeners.remove(sl);
			extendedStatusListeners.remove(sl);
		}
	}
	
	public Order createOrder(OrderListener listener, String isinOrSymbol, String exchange, boolean isSell, int size, double limit, int restrict)
	{
		Order o=Order.newOrder();
		o.addListener(listener);
		o.exchange=exchange;
		o.isSell=isSell;
		o.isin=isinOrSymbol;
		o.symbol=o.isin;
		
		StringBuffer sb=new StringBuffer("AO|SYMBOL=");
		sb.append(isinOrSymbol);
		sb.append("|EXCHANGE=");
		sb.append(exchange);
		sb.append("|BUYSELL=");
		sb.append(isSell?"SELL":"BUY");
		sb.append("|LIMIT=");
		sb.append(limit);
		sb.append("|SIZE=");
		sb.append(size);
		sb.append("|EXTERNID=");
		sb.append(o.externid);
		sb.append("|UPDATE=ON");
		sb.append("|PRICECHECK=ON");
		
		if(restrict==RESTRICT_IOC)
			sb.append("|FILLTYPE=IOC");
		else if(restrict==RESTRICT_FOK)
			sb.append("|FILLTYPE=FOK");
		
		safeWrite(sb.toString());
		return o;
	}
	
	public boolean cancelOrder(Order o)
	{		
		StringBuffer sb=new StringBuffer("OX|");
		if (!o.externid.equals("")) {
			sb.append("EXTERNID=");
			sb.append(o.externid);
		} else if (!o.genid.equals("")){
			sb.append("GENID=");
			sb.append(o.genid);
		} else {
			return false;
		}
		safeWrite(sb.toString());
		return true;
	}
	
	public static final int RESTRICT_NONE=0;
	public static final int RESTRICT_IOC=1;
	public static final int RESTRICT_FOK=2;
	
	/**************************************************************************/
	
	/* Here be dragons */
	
	private String remoteHost;
	private int remotePort;
	
	private List<Quote> quotes=new ArrayList<Quote>();
	private Map<String,Quote> quotesBySymbolExchange=new HashMap<String,Quote>();
	private Map<String,Quote> quotesByISINExchange=new HashMap<String,Quote>();
	
	protected Socket s;
	private PrintWriter sw;
	private BufferedReader sr;
	private APIWriteThread writeThread;
	
	private void safeWrite(String l)
	{
		checkThread();
		synchronized(writeQueue)
		{
			writeQueue.addLast(l);
			writeQueue.notify();
		}
	}
	
	private void resubscribeTickers()
	{
		synchronized(quotes)
		{
			for (Quote q : quotes)
			{
				int l1count = q.getL1Count();
				int l2count = q.getL2Count();
				for (int i=0; i!=(l1count-l2count); i++)
					safeWrite(q.getSubscriptionString(true,false));
				for (int i=0; i!=l2count; i++)
					safeWrite(q.getSubscriptionString(true,true));
			}
		}
	}
	
	private Quote findQ(Map<String,String> parms)
	{
		String isin=parms.get("ISIN");
		String exchange=parms.get("EXCHANGE");
		synchronized(quotes)
		{
			String key=isin+"."+exchange;
			Quote q=quotesByISINExchange.get(key);
			if(q!=null)
				return q;
			String symbol=parms.get("SYMBOL");
			String skey=symbol+"."+exchange;
			q=quotesBySymbolExchange.get(skey);
			if(q!=null)
			{
				// Found by Symbol, store ISIN mapping, too
				q.isin=isin;
				quotesByISINExchange.put(key,q);
				return q;
			}
		}
		return null;
	}
	
	private class APIWriteThread extends Thread
	{
		public APIWriteThread()
		{
			super("tradebase ClientAPI Handler");
			setDaemon(true);
			writeQueue.clear();
		}
		
		@Override
		public void run()
		{
			try
			{
				for(;;)
				{
					String l=null;
					synchronized(writeQueue)
					{
						if(writeQueue.size()>0)
							l=writeQueue.removeFirst();
						else
							writeQueue.wait(10*1000);
					}
					if(l!=null)
					{
						sw.println(l);
					}
					else
					{
						sw.println("E|TEXT="+System.currentTimeMillis());
					}
				}
			}
			catch(Exception e)
			{
				e.printStackTrace();
				// Force closure
				try
				{
					s.close();
				}
				catch(Exception e2)
				{
					// Ignore, we're closing anyway
				}
			}
		}
	}
	
	private class APIThread extends Thread
	{
		public APIThread()
		{
			super("TradeBase MX ClientAPI Handler");
			setDaemon(true);
		}
		
		private void handleNP (Map<String,String> parms ){
			if ( positionListener != null ){
				positionListener.positionChanged( new Position(parms));
			}
		}

		private void handleQTR(Map<String,String> parms)
		{
			String step;
			double defaultStep;
			Quote q=findQ(parms);
			if(q==null)
				return; // hm?
			if(q.symbol==null)
				q.symbol=parms.get("SYMBOL");
			if(q.name==null)
				q.name=parms.get("NAME");
			if(q.curr==null)
				q.curr=parms.get("CURRENCY");
			step = parms.get("PRICESTEP");
			if (step == null || step.isEmpty()) {
				defaultStep = 0.01;
			} else {
				defaultStep = Double.parseDouble(step);
			}
			if (q.getStep() == 0) {
				q.setStep(defaultStep);
			}
			q.fireChanged(QuoteListener.QUOTECHANGED_INFO, 0);
		}
		
		private double gp(Map<String,String> parms)
		{
			try
			{
				return Double.parseDouble(parms.get("PRICE"));
			}
			catch(Exception e)
			{
				return 0;
			}
		}
		
		private int gs(Map<String,String> parms)
		{
			try
			{
				return Integer.parseInt(parms.get("SIZE"));
			}
			catch(Exception e)
			{
				return 0;
			}
		}
		
		private void handleTick(Map<String,String> parms) throws NumberFormatException
		{
			Quote q=findQ(parms);
			if(q==null)
				return; // hm?

			String phase=parms.get("PHASE");
			if(phase!=null && !phase.equals(q.phase))
			{
				q.phase=phase;
			}
			else
				phase=null;
			String time = parms.get("TIME");
			if (time != null) {
				q.time = time;
			}
			
			String type=parms.get("TYPE");
			if("LAST".equals(type))
			{
				q.last=gp(parms);
				q.lastsize=gs(parms);
				q.fireChanged(QuoteListener.QUOTECHANGED_LAST, 0);
			}
			else if("BID".equals(type))
			{
				q.bid=gp(parms);
				q.bidsize=gs(parms);
				q.fireChanged(QuoteListener.QUOTECHANGED_BID, 0);
			}
			else if("ASK".equals(type))
			{
				q.ask=gp(parms);
				q.asksize=gs(parms);
				q.fireChanged(QuoteListener.QUOTECHANGED_ASK, 0);
			}
			else if("L2BID".equals(type) || "L2ASK".equals(type))
			{
				double[] prices;
				int[] sizes;
				int changed;
				if ("L2BID".equals(type))
				{
					prices = q.l2bid;
					sizes = q.l2bidsize;
					changed = QuoteListener.QUOTECHANGED_L2BID;
				}
				else
				{
					prices = q.l2ask;
					sizes = q.l2asksize;
					changed = QuoteListener.QUOTECHANGED_L2ASK;
				}
				int rank = Integer.parseInt(parms.get("RANK"));
				if (rank<0 || rank>=prices.length)
					return;
				prices[rank] = gp(parms);
				sizes[rank] = gs(parms);
				q.fireChanged(changed, rank);
			}
			if(phase!=null)
			{
				q.fireChanged(QuoteListener.QUOTECHANGED_PHASE, 0);
			}
		}
		
		private void handleOS(Map<String,String> parms)
		{
			String externid=parms.get("EXTERNID");
			if(externid!=null)
			{
				Order o=Order.findOrder(externid);
				if(o!=null)
					o.handleOS(parms);
			}
		}
		
		private void handleE(Map<String,String> parms)
		{
			String text=parms.get("TEXT");
			if(text!=null)
			{
				try
				{
					long ts=System.currentTimeMillis()-Long.parseLong(text);
					fireAPILatencyChanged(ts);
				}
				catch(NumberFormatException nfe)
				{
					// Not a ping reply, ignore
				}
			}
		}
		
		private void doIt() throws Exception
		{
			s=new Socket(remoteHost,remotePort);
			sw=new PrintWriter(s.getOutputStream(),true);
			sr=new BufferedReader(new InputStreamReader(s.getInputStream()));

			writeThread=new APIWriteThread();
			writeThread.start();
			
			// Send Client Information
			sw.println("CI|REV=2.0|NAME=Java-ClientAPI");
			
			if ( enablePositionUpdate){
				sw.println("NPR|UPDATE=ON");
			}
			else{
				sw.println("NPR|UPDATE=OFF");
			}
			
			fireStatusChanged(true);
			
			resubscribeTickers();
			
			for(;;)
			{
				String l=sr.readLine();
				
				if(l==null)
					throw new RuntimeException("Closed");
				StringTokenizer stz=new StringTokenizer(l,"|");
				String cmd=stz.nextToken();
				// Split params
				Map<String,String> parms=new HashMap<String,String>();
				while(stz.hasMoreTokens())
				{
					String tok=stz.nextToken();
					StringTokenizer splitter=new StringTokenizer(tok,"=");
					String key=splitter.nextToken();
					String val=null;
					if(splitter.hasMoreTokens())
						val=splitter.nextToken();
					parms.put(key,val);
				}
				
				if("T".equals(cmd))
					handleTick(parms);
				else if("QTR".equals(cmd))
					handleQTR(parms);
				else if("OS".equals(cmd))
					handleOS(parms);
				else if("E".equals(cmd))
					handleE(parms);
				else if ( "NP".equals(cmd)){
					handleNP(parms);
				}
			}
		}
		
		@Override
		public void run()
		{
			for(;;)
			{
				try
				{
					doIt();
				}
				catch(Exception e)
				{
					e.printStackTrace();
					try
					{
						s.close();
						writeThread.interrupt();
					}
					catch(IOException ioe)
					{
						// Ignore, we're closing
					}
				}
				fireStatusChanged(false);
				try
				{
					Thread.sleep(3000);
				}
				catch (InterruptedException ie)
				{
					// Don't care too much - the pause between connection attempts will just be shorter
				}
			}
		}
	}
	
	private APIThread instance=null;
	private LinkedList<String> writeQueue=new LinkedList<String>();
	private synchronized void checkThread()
	{
		if(instance==null)
		{
			instance=new APIThread();
			instance.start();
		}
	}

	private List<StatusListener> statusListeners=new ArrayList<StatusListener>();
	private List<ExtendedStatusListener> extendedStatusListeners=new ArrayList<ExtendedStatusListener>();
	private void fireStatusChanged(boolean newstatus)
	{
		StatusListener sl[];
		synchronized(statusListeners)
		{
			if(statusListeners.size()==0)
				return;
			sl=statusListeners.toArray(new StatusListener[statusListeners.size()]);
		}
		for(int c=0;c<sl.length;c++)
			sl[c].apiStatusChanged(newstatus);
	}
	
	private void fireAPILatencyChanged(long newLatency)
	{
		ExtendedStatusListener sl[];
		synchronized(statusListeners)
		{
			if(extendedStatusListeners.size()==0)
				return;
			sl=extendedStatusListeners.toArray(new ExtendedStatusListener[extendedStatusListeners.size()]);
		}
		for(int c=0;c<sl.length;c++)
		{
			sl[c].APILatency(newLatency);
		}
	}
}
